package com.chianghao.nio.core;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author ChiangHao
 * @date 2022/10/2 10:20
 * @Description:
 */
public class NioReactor implements Runnable {


    private final ServerSocketChannel servSocketChannel;
    private final Selector selector;

    public NioReactor(String ip, int port) throws IOException {
        servSocketChannel = ServerSocketChannel.open();
        servSocketChannel.configureBlocking(false);
        //绑定端口
        if (ip == null || ip.length() == 0) {
            servSocketChannel.socket().bind(new InetSocketAddress(port),50_000);
        } else {
            servSocketChannel.socket().bind(new InetSocketAddress(ip, port),50_000);
        }
        //Select 创建
        selector = Selector.open();
        //向Selector 注册通道（用户监控单一事件）
        SelectionKey key = servSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    private final static LinkedBlockingQueue<Runnable> executorQueue = new LinkedBlockingQueue<>();
    private final static ExecutorService executor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors()*3,
            60L,
            TimeUnit.SECONDS,
            executorQueue,
            r -> new Thread(r,"分发线程")
    );

    @Override
    public void run() {
        List<NioSubReactor> subReactors = NioService.getSubReactors();
        int index = 0;
        int length = subReactors.size()-1;
        int count = 0;
        for (; ; ) {
            try {
                selector.select();
            } catch (IOException e) {
                e.printStackTrace();
            }
            Iterator<SelectionKey> it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next();
                if(key.isValid()){
                    if(key.isAcceptable()){
                        System.out.println("接收===>"+(++count));
                        SocketChannel channel = null;
                        try {
                            channel = servSocketChannel.accept();
                        } catch (IOException e) {
                            e.printStackTrace();
                            try {
                                channel.close();
                            } catch (IOException ex) {
                                ex.printStackTrace();
                            }
                        }
                        if (channel != null) {
                            // 通过线程分发
                            executor.execute(new Dispatcher(channel,subReactors.get(index)));
                            if (index >= length) {
                                index = 0;
                            } else {
                                index++;
                            }
                        }
                    }
                }
                it.remove();
            }
        }
    }

    static class Dispatcher implements Runnable{

        SocketChannel channel;
        NioSubReactor subReactor;

        public Dispatcher(SocketChannel channel,NioSubReactor subReactor){
            this.subReactor = subReactor;
            this.channel = channel;
        }

        @Override
        public void run() {
            subReactor.register(channel);
        }
    }

}
